package com.zyh.kafka.day03;

import com.zyh.kafka.util.KafkaConsumerUtils;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class KafkaConsumerTest3 {
    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = KafkaConsumerUtils.get("g3");

        try {
            while(true){
                ConsumerRecords<String, String> records = consumer.poll(1000);
                Set<TopicPartition> partitions = records.partitions();
                for (TopicPartition partition : partitions) {
                    System.out.println("消费了当前分区：" + partition);
                    List<ConsumerRecord<String, String>> records1 = records.records(partition);
                    for (ConsumerRecord<String, String> record : records1) {
                        System.out.println(record);
                    }
                    Map<TopicPartition, OffsetAndMetadata> offSet = new HashMap<>();
                    offSet.put(partition,new OffsetAndMetadata(records1.get(records1.size()-1).offset()+1));
                    //同步提交-提交一个分区的位移
                    //consumer.commitSync(offSet);

                    //异步提交-回调函数
                    consumer.commitAsync(offSet, new OffsetCommitCallback() {
                        /**
                         *
                         * @param map 表示本次提交的消费位移
                         * @param e 提交失败的异常
                         */
                        @Override
                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                            System.out.println("map = " + map);
                            System.out.println("e = " + e);
                        }
                    });
                    System.out.println("提交了"+ partition + "的消费位移");
                }
                //同步提交-手动提交消费位移
                //consumer.commitSync();//将当前消费者对各个分区的消费位移全部提交

                //异步提交
                //consumer.commitAsync();//异步的将当前消费者对各个分区的消费位移全部提交

                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                        System.out.println("map = " + map);
                        System.out.println("e = " + e);
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            KafkaConsumerUtils.close(consumer);
        }
    }
}
